package com.example.config;

import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class FlinkEnvConfig {
    private final FlinkConfig flinkConfig;

    public FlinkEnvConfig(FlinkConfig flinkConfig) {
        this.flinkConfig = flinkConfig;
    }

    @Bean
    public StreamExecutionEnvironment streamExecutionEnvironment() throws Exception {
        org.apache.flink.configuration.Configuration configuration = new org.apache.flink.configuration.Configuration();
        configuration.setString(RestOptions.ADDRESS, flinkConfig.getRemote().getHost());
        configuration.setInteger(RestOptions.PORT, flinkConfig.getRemote().getPort());

        StreamExecutionEnvironment env = StreamExecutionEnvironment.createRemoteEnvironment(
                flinkConfig.getRemote().getHost(),
                flinkConfig.getRemote().getPort(),
                configuration
        );
        env.enableCheckpointing(flinkConfig.getJob().getCheckpointInterval());
        env.setParallelism(flinkConfig.getJob().getParallelism());
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5000));
        return env;
    }
}
